#pragma once 

#include <memory>
#include <atomic>
#include <array>
#include <list>

#include <boost/asio/co_spawn.hpp>
#include <boost/asio/detached.hpp>
#include <boost/asio/use_awaitable.hpp>
#include <boost/asio/redirect_error.hpp>
#include <boost/asio/experimental/channel.hpp>
#include <boost/asio/experimental/concurrent_channel.hpp>

#include "server/http/http_conn.hpp"
#include "protocol/http/http_parser.hpp"
#include "protocol/http/http_define.h"
#include "protocol/http/http_request.hpp"
#include "core/session.hpp"
#include "protocol/http/http_response.hpp"

namespace mms{
template<typename CONN_TYPE>
class HttpRequestHandler{
public:
    virtual boost::asio::awaitable<bool> on_new_request(
        std::shared_ptr<HttpServerSession<CONN_TYPE>> session,
        std::shared_ptr<HttpRequest> req,
        std::shared_ptr<HttpResponse> resp
    ) = 0;
};

template<typename CONN_TYPE>
class HttpServerSession: public Session{
public:
    HttpServerSession(HttpRequestHandler<CONN_TYPE>* request_handler,
     std::shared_ptr<CONN_TYPE> conn):Session(conn->get_worker()),
        request_handler_(request_handler),conn_(conn),
        recv_coroutine_exited_(get_worker()->get_io_context()){
            set_session_type("http");
        }
    
    virtual ~HttpServerSession(){
        spdlog::info("dsetroy HttpServerSession");
    }

    void service(){
        auto self(shared_from_this());
        boost::asio::co_spawn(conn_->get_worker()->get_io_context(),[this,self]()
        ->boost::asio::awaitable<void>{
            recv_coroutinr_running_ = true;
            http_parser_.on_http_request(std::bind(&HttpServerSession<CONN_TYPE>::on_http_request,
            this,std::placeholders::_1));
            co_await conn_->cycle_recv([this](const char* buf, size_t len)
                ->boost::asio::awaitable<std::pair<bool,int32_t>>{
                    int32_t total_consumed = 0;
                    int32_t consumed = 0;
                    do{
                        consumed = co_await http_parser_.read(std::string_view(buf + total_consumed,len));
                        if(consumed < 0){
                            co_return std::make_pair(false,-1);
                        }
                        total_consumed += consumed;
                        len -= consumed;
                    }while(consumed > 0 && len > 0);
                    co_return std::make_pair(true,total_consumed);
                });
                co_return;
        },[this,self](std::exception_ptr exp){
            (void)exp;
            recv_coroutine_exited_.close();
            recv_coroutinr_running_ = false;
        });
    }

    void close(){
        if(closed_.test_and_set()){
            return;
        }
        auto self(shared_from_this());
        boost::asio::co_spawn(get_worker()->get_io_context(), [this, self]()
        ->boost::asio::awaitable<void>{
            boost::system::error_code ec;
            if(conn_){
                conn_->close();
            }
            if(recv_coroutinr_running_){
                co_await recv_coroutine_exited_.async_receive(boost::asio::redirect_error(
                    boost::asio::use_awaitable,ec
                ));
            }
            conn_.reset();
            co_return;
        }, boost::asio::detached);
    }

    boost::asio::awaitable<void> on_http_request(std::shared_ptr<HttpRequest> req){
        if(curr_req_){
            reqs_.push_back(req);
            co_return;
        }
        curr_req_ = req;
        co_await process_one_req(curr_req_);
        co_return;
    }

    boost::asio::awaitable<void> process_one_req(std::shared_ptr<HttpRequest> req){
        auto self(shared_from_this());
        boost::asio::co_spawn(get_worker()->get_io_context(),[this,self,req]()
            ->boost::asio::awaitable<void>{
                std::shared_ptr<HttpResponse> resp = std::make_shared<HttpResponse>(conn_);
                co_await request_handler_->on_new_request
                    (std::static_pointer_cast<HttpServerSession<CONN_TYPE>>(shared_from_this()),
                    req,resp);
                co_return;
            },boost::asio::detached);
            co_return;
    }
    

    boost::asio::awaitable<void> close_or_do_next_req() {
        if (!curr_req_) {
            co_return;
        }

        if (curr_req_->get_header("Connection") != "keep-alive") {// 非长连接，关闭
            conn_->close();
            co_return;
        }

        // 必然在最前面，去掉最前面的请求
        if (reqs_.size() <= 0) {
            curr_req_ = nullptr;
            co_return;
        }

        curr_req_ = reqs_.front();
        reqs_.pop_front();

        co_await process_one_req(curr_req_);
        co_return;
    }

    ThreadWorker* get_worker() {
        return conn_->get_worker();
    }

protected:
    std::atomic_flag closed_ = ATOMIC_FLAG_INIT;
    HttpRequestHandler<CONN_TYPE>* request_handler_;
    std::shared_ptr<CONN_TYPE> conn_;
    HttpParser http_parser_;
    std::shared_ptr<HttpRequest> curr_req_;
    std::list<std::shared_ptr<HttpRequest>> reqs_;
    bool recv_coroutinr_running_ = false;
    boost::asio::experimental::concurrent_channel<void(boost::system::error_code,bool)>
        recv_coroutine_exited_;
};


}

